package org.apache.rocketmq.proxy.processor;

import io.netty.channel.Channel;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.proxy.common.Address;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.service.metadata.MetadataService;
import org.apache.rocketmq.proxy.service.relay.ProxyRelayService;
import org.apache.rocketmq.proxy.service.route.ProxyTopicRouteData;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;

import java.time.Duration;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

public interface MessagingProcessor extends StartAndShutdown {

    long DEFAULT_TIMEOUT_MILLS = Duration.ofSeconds(2).toMillis();


    CompletableFuture<RemotingCommand> request(ProxyContext ctx, String brokerName, RemotingCommand request,
                                               long timeoutMillis);

    CompletableFuture<Void> requestOneway(ProxyContext ctx, String brokerName, RemotingCommand request,
                                          long timeoutMillis);

    ProxyTopicRouteData getTopicRouteDataForProxy(ProxyContext context, List<Address> addressList, String topic) throws Exception;

    MetadataService getMetadataService();

    default CompletableFuture<List<SendResult>> sendMessage(
            ProxyContext ctx,
            QueueSelector queueSelector,
            String producerGroup,
            int sysFlag,
            List<Message> msg
    ) {
        return sendMessage(ctx, queueSelector, producerGroup, sysFlag, msg, DEFAULT_TIMEOUT_MILLS);
    }

    CompletableFuture<List<SendResult>> sendMessage(
            ProxyContext ctx,
            QueueSelector queueSelector,
            String producerGroup,
            int sysFlag,
            List<Message> msg,
            long timeoutMillis
    );

    SubscriptionGroupConfig getSubscriptionGroupConfig(
            ProxyContext ctx,
            String consumerGroupName
    );

    void registerProducer(
            ProxyContext ctx,
            String producerGroup,
            ClientChannelInfo clientChannelInfo
    );

    void unRegisterProducer(
            ProxyContext ctx,
            String producerGroup,
            ClientChannelInfo clientChannelInfo
    );

    Channel findProducerChannel(
            ProxyContext ctx,
            String producerGroup,
            String clientId
    );

    void registerProducerListener(
            ProducerChangeListener producerChangeListener
    );

    void registerConsumer(
            ProxyContext ctx,
            String consumerGroup,
            ClientChannelInfo clientChannelInfo,
            ConsumeType consumeType,
            MessageModel messageModel,
            ConsumeFromWhere consumeFromWhere,
            Set<SubscriptionData> subList,
            boolean updateSubscription
    );

    ClientChannelInfo findConsumerChannel(
            ProxyContext ctx,
            String consumerGroup,
            Channel channel
    );

    void unRegisterConsumer(
            ProxyContext ctx,
            String consumerGroup,
            ClientChannelInfo clientChannelInfo
    );

    void registerConsumerListener(
            ConsumerIdsChangeListener consumerIdsChangeListener
    );

    void doChannelCloseEvent(String remoteAddr, Channel channel);

    ConsumerGroupInfo getConsumerGroupInfo(ProxyContext ctx, String consumerGroup);

    ProxyRelayService getProxyRelayService();
}

